#pragma once
#include "detail.hpp"
#include "files.hpp"
#include "abstract.hpp"
#include "message.hpp"
#include <muduo/net/TcpServer.h>
#include <muduo/net/TcpClient.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/EventLoopThread.h>
//#include <thread>
#include <mutex>
#include <utility>

namespace rpc
{
    class MuduoBuffer : public BaseBuffer
    {
     public:
        using ptr = std::shared_ptr<MuduoBuffer>; 
        MuduoBuffer(muduo::net::Buffer* buff):_buff(buff)
        {}
        virtual size_t readableSize() override
        {
            return _buff->readableBytes();
        }
        virtual int32_t peekInt32() override
        {
            //会进行网络字节序的转换
            return _buff->peekInt32();
        }
        virtual void retrieveInt32() override
        {
            return _buff->retrieveInt32();
        }
        virtual int32_t readInt32() override
        {
            return _buff->readInt32();
        }
        virtual std::string retrieveAsString(size_t len) override
        {
            return _buff->retrieveAsString(len);
        }
    private:
        muduo::net::Buffer* _buff;
    };

    class BufferFactory
    {
    public:
        template<typename ...Args>
        static BaseBuffer::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    class LVProtocol : public BaseProtocol
    {
    public:
        // [len][value]
        // [len][mtype][idlen][id][body]
        using ptr = std::shared_ptr<LVProtocol>;
        //判断缓冲区的数据量是否足够一条消息的处理
        virtual bool canProcessed(const BaseBuffer::ptr& buf) override
        {
            if(buf->readableSize() <= sizeof(int32_t))
            {
                DLOG("数据不足4字节");
                return false;
            }
            if(buf->peekInt32() < lenFieldsLength)
            {
                DLOG("数据量不足！");
                return false;
            }
            //取出4字节的数据，该数据根据协议是总长度
            int32_t totol_len = buf->peekInt32();
            if((totol_len + lenFieldsLength) > buf->readableSize())
                return false;
            return true;
        }
        virtual bool onMessage(const BaseBuffer::ptr& buf, BaseMessage::ptr& msg) override
        {
            int32_t total_len = buf->readInt32(); //读取总长度
            MType mtype = (MType)buf->readInt32(); //读取数据类型
            int32_t id_len = buf->readInt32(); //读取id长度
            int32_t body_len = total_len - id_len - idlenFieldsLength - mtypeFieldsLength;
            std::string id = buf->retrieveAsString(id_len);
            std::string body = buf->retrieveAsString(body_len);

            msg = MessageFactory::create(mtype);
            if(!msg.get())
            {
                ELOG("消息类型错误！");
                return false;
            }
            if(!msg->deserialize(body))
            {
                ELOG("消息正文反序列化失败！");
                return false;
            }
            
            msg->setId(id);
            msg->setMtype(mtype);
            return true;
        }
        virtual std::string serialize(const BaseMessage::ptr& msg) override
        {
            // [len][mtype][idlen][id][body]
            std::string body = msg->serialize();
            std::string id = msg->rid();
            int32_t id_len = htonl(id.size());
            auto mtype = htonl((int32_t)msg->mtype());
            int32_t h_total_len = body.size() + id.size() + idlenFieldsLength + mtypeFieldsLength;
            int32_t n_total_len = htonl(h_total_len);
            std::string ret;
            ret.reserve(h_total_len);
            ret.append((char*)&n_total_len, lenFieldsLength);
            ret.append((char*)&mtype, mtypeFieldsLength);
            ret.append((char*)&id_len, idlenFieldsLength);
            ret.append(id);
            ret.append(body);

            return ret;
        }
    private:
        const size_t lenFieldsLength = 4;
        const size_t mtypeFieldsLength = 4;
        const size_t idlenFieldsLength = 4;
    };

    class ProtocolFactory
    {
    public:
        template<typename ...Args>
        static BaseProtocol::ptr create(Args&& ...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };

    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;
        MuduoConnection(const muduo::net::TcpConnectionPtr& conn, const BaseProtocol::ptr& protocol):
        _protocol(protocol), _conn(conn)
        {}
        virtual void send(const BaseMessage::ptr& msg) override
        {
            std::string body = _protocol->serialize(msg);
            _conn->send(body);
        }
        virtual void shutdown() override
        {
            _conn->shutdown();
        }
        virtual bool connected() override
        {
            return _conn->connected();
        }
    private:
        BaseProtocol::ptr _protocol;
        muduo::net::TcpConnectionPtr _conn;
    };

    class ConnectionFactory
    {
    public:
        template<typename ...Args>
        static BaseConnection::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    class MuduoServer : public BaseServer
    {
    public:
        using ptr = std::shared_ptr<MuduoServer>;
        MuduoServer(int port):_server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "MuduoServer", muduo::net::TcpServer::kReusePort),
        _protocol(ProtocolFactory::create())
        {}
        virtual void start()
        {
            _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&MuduoServer::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _server.start(); //开始事件监听
            _baseloop.loop(); //开启死循环事件监控
        }
    private:
        void onConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if(conn->connected())
            {
                std::cout<<"建立连接！"<<std::endl;
                auto muduo_conn = ConnectionFactory::create(conn, _protocol);
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns[conn] = muduo_conn;
                }
                if(_cb_connection) _cb_connection(muduo_conn);
            }
            else
            {
                std::cout<<"连接断开"<<std::endl;
                BaseConnection::ptr muduo_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if(it == _conns.end())
                    {
                        return;
                    }
                    muduo_conn = it->second;
                    _conns.erase(conn);
                    if(_cb_close)     
                    {
                        _cb_close(muduo_conn);
                    }
                }

            }
        }
        void onMessage(const muduo::net::TcpConnectionPtr& conn, muduo::net::Buffer* buff, muduo::Timestamp)
        {
            //std::lock_guard<std::mutex> lock(_mutex);
            auto base_buff = BufferFactory::create(buff);
            while(1)
            {
                if(!_protocol->canProcessed(base_buff))
                {
                    //数据不足
                    if(base_buff->readableSize() > maxDateSize)
                    {
                        conn->shutdown();
                        ELOG("缓冲区中数据过大！");
                        return;
                    }
                    break;
                }
                BaseMessage::ptr msg;
                //protocol的onmessage包含反序列化
                bool ret = _protocol->onMessage(base_buff, msg);
                if(!ret)
                {
                    conn->shutdown();
                    ELOG("缓冲区中数据错误");
                    return;
                }
                BaseConnection::ptr base_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if(it == _conns.end())
                    {
                        conn->shutdown();
                        return;
                    }
                    base_conn = it->second;
                }
                if(_cb_message)  _cb_message(base_conn, msg);
            }
        }
    private:
        const size_t maxDateSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;
        std::mutex _mutex;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
    };

    class ServerFactory
    {
    public:
        template<typename ...Args>
        static BaseServer::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

     class MuduoClient : public BaseClient
    {
    public:
        using ptr = std::shared_ptr<BaseClient>;
        MuduoClient(const std::string& sip, int sport):_baseloop(_loopthread.startLoop()),
        _downlatch(1),
        _client(_baseloop, muduo::net::InetAddress(sip, sport), "MuduoClient"),
        _protocol(ProtocolFactory::create())
        {}
        virtual void connect() override
        {
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this, 
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));

            //连接服务器
            _client.connect();
            _downlatch.wait();
        }
        virtual void shutdown() override
        {
            _client.disconnect();
        }
        virtual bool send(const BaseMessage::ptr& msg) override
        {
            std::lock_guard<std::mutex> lock(_mutex);
            if(!connected() || !_conn)
            {
                ELOG("连接已断开！");
                return false;
            }
            else
            {
                _conn->send(msg);
                return true;
            }

        }
        virtual BaseConnection::ptr connection() override
        {
            return _conn;
        }
        virtual bool connected() override
        {
            return (_conn && _conn->connected());
        }
    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if(conn->connected())
            {
                std::cout << "建立连接！" << std::endl;
                _downlatch.countDown(); //计数--， 为0时唤醒阻塞
                _conn = ConnectionFactory::create(conn, _protocol);
            }
            else
            {
                std::cout << "连接断开！" << std::endl;
                _conn.reset();
            }
        }
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buff, muduo::Timestamp)
        {
            //std::lock_guard<std::mutex> lock(_mutex);
            auto base_buff = BufferFactory::create(buff);
            while(1)
            {
                if(!_protocol->canProcessed(base_buff))
                {
                    //数据不足
                    if(base_buff->readableSize() > maxDateSize)
                    {
                        conn->shutdown();
                        ELOG("缓冲区中数据过大！");
                        return;
                    }
                    break;
                }
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(base_buff, msg);
                if(!ret)
                {
                    conn->shutdown();
                    ELOG("缓冲区中数据错误");
                    return;
                }
                if(_cb_message)  _cb_message(_conn, msg);
            }
        }
    private:
        std::mutex _mutex;
        const size_t maxDateSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        BaseConnection::ptr _conn;
        muduo::CountDownLatch _downlatch;
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop *_baseloop;
        muduo::net::TcpClient _client;
    };

    class ClientFactory
    {
    public:
        template<typename ...Args>
        static BaseClient::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };
}


